
package com.bangcommunity.bbframe.mq.consumer;

import com.bangcommunity.bbframe.common.base.MainConfigLoader;
import com.bangcommunity.bbframe.common.utils.io.PropertiesUtil;
import com.bangcommunity.bbframe.common.utils.lang.SpiUtils;
import com.bangcommunity.bbframe.common.utils.lang.StringUtils;
import com.bangcommunity.bbframe.dynconfig.DynConfigRegister;
import com.bangcommunity.bbframe.dynconfig.IDynConfigProcessor;
import com.bangcommunity.bbframe.dynconfig.parser.ConfigValueParser;
import com.bangcommunity.bbframe.dynconfig.parser.StringConfigValueParser;
import com.bangcommunity.bbframe.mq.IQtMqFactory;
import com.bangcommunity.bbframe.mq.kafka.producer.KafkaProducerClient;
import com.bangcommunity.bbframe.mq.producer.IQtProducer;
import org.slf4j.Logger;
import org.slf4j.LoggerFactory;

import java.util.Properties;

/**
 * Created by tanghc on 17/1/20.
 */
public class QtConsumerClient {

    static Logger logger = LoggerFactory.getLogger(KafkaProducerClient.class);

    private QtConsumerClient() {
    }

    private static class QtProducerClientHolder {

        private static IQtProducer instance;
        private static final String mq_config_group = "common";
        private static final String mq_config_dataId = "mqConfig";
        private static String mq = "kafka";
        private static boolean isInit = false;
        private static final String productCode = MainConfigLoader.getInstance().getProductCode();
        private static final String appCode = MainConfigLoader.getInstance().getAppCode();
        static {
            DynConfigRegister.getInstance().buildOnce(productCode, appCode, mq_config_group, mq_config_dataId,
                    new IDynConfigProcessor() {

                        @Override
                        public void procesConfig(String config) {

                            Properties properties = PropertiesUtil.loadFromString(config);
                            ConfigValueParser<String> configValueParser = new StringConfigValueParser();
                            mq = configValueParser.parse(properties, "consumer", "kafka");
                        }
                    });
        }

        private static Runnable builderConsumer(String rawTopic, Properties properties, IQtMsgProcessor msgProcessor) {

            IQtMqFactory mqFactory = SpiUtils.getSpiBean(mq, IQtMqFactory.class);
            if (null != mqFactory) {
                return mqFactory.builderConsumer(rawTopic, properties, msgProcessor);
            } else {
                throw new RuntimeException("no qtmqfactory spi:" + mq);
            }
        }

    }

    public static Runnable builderConsumer(String rawTopic, IQtMsgProcessor msgProcessor) {

        return builderConsumer(rawTopic, null, msgProcessor);
    }

    public static Runnable builderConsumer(String projectCode, String appCode, String topic,
            IQtMsgProcessor msgProcessor) {

        return builderConsumer(projectCode, appCode, topic, null, msgProcessor);
    }

    public static Runnable builderConsumer(String rawTopic, Properties properties, IQtMsgProcessor msgProcessor) {
        return QtProducerClientHolder.builderConsumer(rawTopic, properties, msgProcessor);
    }

    public static Runnable builderConsumer(String projectCode, String appCode, String topic, Properties properties,
            IQtMsgProcessor msgProcessor) {
        return builderConsumer(StringUtils.join(new Object[] { projectCode, appCode, topic }, '_'), properties,
                msgProcessor);
    }
}
